package jp.aist.gtrc.plus.scheduler.algorithm;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.*;
import java.sql.*;
import javax.sql.*;

import jp.aist.gtrc.plus.admission.AdmissionOperationManager;
import jp.aist.gtrc.plus.admission.AdmissionServer;
import jp.aist.gtrc.plus.admission.QcommandClient;
import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSJobStatus;
import jp.aist.gtrc.plus.scheduler.specific.pbs.status.PBSNodeStatus;
import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.status.NodeStatus;
import jp.aist.gtrc.plus.scheduler.status.StatusManager;

import org.apache.log4j.Logger;

/*
 * Job Validator terminates jobs missing the deadline, clear 
 *   unconfirmed jobs, and remove jobs exceeding weighted walltime.
 *
 */
public class JobValidator implements QcommandClient {

	protected static final Logger logger = Logger.getLogger(JobValidator.class);

        private AdmissionOperationManager admissionOperationMgr;
        private StatusManager statusMgr;
        private Collection<String> invalidJobs;
   
        private String dbUrl = null;
        private String dbUser = null;
        private String dbPassword = null;
        private Connection dbConn = null;

        public JobValidator(StatusManager statusMgr, AdmissionServer admissionServer) {
           assert(admissionServer != null);
           logger.debug("JobValidator constructor()");
           this.statusMgr = statusMgr;
           this.admissionOperationMgr = (AdmissionOperationManager) admissionServer;

           try {
              String home = System.getProperty("user.home");
              String dbConf = home + "/mysqldb.conf";
              System.out.println("Reading mysqldb.conf at " + home);
              InputStream is = new FileInputStream(dbConf);
  
              Properties prop = new Properties();
              prop.load(is); 

              this.dbUrl = "jdbc:mysql://" + prop.getProperty("DBHOST") + ":" + prop.getProperty("DBPORT") + "/" + prop.getProperty("DBNAME"); 
              this.dbUser = prop.getProperty("MYSQL_USER"); 
              this.dbPassword = prop.getProperty("MYSQL_PWD"); 
              System.out.println(this.dbUrl);

              Class.forName("com.mysql.jdbc.Driver");
              Connection dbConn = getDbConnection();
              Statement dbStmt = dbConn.createStatement();
              if (dbStmt != null) {
                 dbStmt.execute("show tables"); // test mysql 
              }
              dbConn.close();
           } catch (Exception e) {
              e.printStackTrace();
              System.err.println("JobValidator cannot connect to database");
              System.exit(1);
           }
        }

        /*
         * Just return Torque's job id (integer) excluding array id and 
         *   host parts.
         */
        private String getCanonicalJobId(String jobId) {
           String canonicalJobId = jobId;
           canonicalJobId = jobId.substring(0,jobId.indexOf('.'));
           if (canonicalJobId.indexOf('-') > 0) {
              System.err.println("Can only work with new Torque version that has qdel supporting job array.");
              System.exit(1);
           }
           if (canonicalJobId.indexOf('[') > 0) { // working with job array
              canonicalJobId = canonicalJobId.substring(0,canonicalJobId.indexOf('[')); 
          
           }
           return canonicalJobId; 
        }

        private Connection getDbConnection() throws Exception {
           if ( (this.dbConn != null) && (!this.dbConn.isClosed()) ) {
              return this.dbConn;
           }

           try {
              Connection dbConn = DriverManager.getConnection(dbUrl,dbUser,dbPassword);
              this.dbConn = dbConn;
              return dbConn;
           } catch (Exception e) {
              e.printStackTrace();
              System.err.println("JobValidator cannot connect to database");
           } 
           return null;
        }


        /*
         * Take care only deadline QoS job
         */
        public void validate() throws Exception {
           System.out.println("validate() queue");
           Collection<NodeStatus> allNodes = statusMgr.getNodes();
           Collection<JobStatus> allJobs = statusMgr.getJobs();

           invalidJobs = new HashSet<String>();
           Set<String> exceedingWeightedWalltimeJobs = new HashSet<String>();
           Set<String> missingDeadlineJobs = new HashSet<String>();
           Set<String> unconfirmedJobs = new HashSet<String>();
           Set<String> noWalltimeJobs = new HashSet<String>();

           int now = (int) (Calendar.getInstance().getTimeInMillis()/1000);


           for (NodeStatus nodeStatus : allNodes) {
              if (!nodeStatus.isAlive()) { // node fails
                 continue;
              }
              double performanceIndex = 1.0;
              String pfi = nodeStatus.getAttribute("pfi");
              if (pfi != null) {
                 performanceIndex = Double.parseDouble(nodeStatus.getAttribute("pfi"));
              }
              Collection<String> jobIDs = ((PBSNodeStatus) nodeStatus).getRunningJobIDs();
              if (jobIDs == null) continue;

              for (String jobID : jobIDs) {
                 JobStatus jobStatus = statusMgr.getJob(jobID);
                 if (jobStatus == null) continue;
                 PBSJobStatus pbsJobStatus = (PBSJobStatus) jobStatus;
                 if (!pbsJobStatus.isQoSJob()) continue; 
                 int estExecTime = pbsJobStatus.getEstimatedExecutionTime();
                 int usedTime = now - (int) (pbsJobStatus.getExecStartTime().getTimeInMillis()/1000);
                 if ((double) usedTime >= estExecTime/performanceIndex) {
                    String jobId = pbsJobStatus.getJobID().toString();
                    exceedingWeightedWalltimeJobs.add(getCanonicalJobId(jobID));
                 } 
              }
           }

           for (JobStatus jobStatus : allJobs) {
              PBSJobStatus pbsJobStatus = (PBSJobStatus) jobStatus;
              if (pbsJobStatus.isCompleted()) continue;
              if (!pbsJobStatus.isQoSJob()) continue;
              if (pbsJobStatus.getDeadline() <= now) { // missing deadline jobs
                 String jobId = pbsJobStatus.getJobID().toString();
                 missingDeadlineJobs.add(getCanonicalJobId(jobId));
              } else if (pbsJobStatus.getEstimatedExecutionTime() < 0) { // no estimated execution time 
                 String jobId = pbsJobStatus.getJobID().toString();
                 noWalltimeJobs.add(getCanonicalJobId(jobId));
              } else if (pbsJobStatus.isHeld()) { // unconfirmed jobs
                 int latestConfirmationTime = (int) (pbsJobStatus.getSubmitTime().getTimeInMillis()/1000) + AdmissionOperationManager.MAX_CONFIRMATION_TIME;
                 //System.out.println(latestConfirmationTime);
                 if (latestConfirmationTime <= now) {
                    String jobId = pbsJobStatus.getJobID().toString();
                    unconfirmedJobs.add(getCanonicalJobId(jobId));
                 }
              }
           }

           if (exceedingWeightedWalltimeJobs.size() > 0) {
              System.out.print("Purge exceeding weighted walltime jobs : ");
              for (String exceedingWeightedWalltimeJob : exceedingWeightedWalltimeJobs) {
                 System.out.print(exceedingWeightedWalltimeJob + " ");
              }
              System.out.println();
              invalidJobs.addAll(exceedingWeightedWalltimeJobs);
           }

           if (missingDeadlineJobs.size() > 0) {
              System.out.print("Purge missing deadline jobs : ");
              for (String missingDeadlineJob : missingDeadlineJobs) {
                 System.out.print(missingDeadlineJob + " ");
                 Connection dbConn = getDbConnection();
                 Statement dbStmt = dbConn.createStatement();
                 if (dbStmt != null) {
                    dbStmt.execute("update jobinfo set jobstate='QOS_VIOLATED' where pbsjobid=" + missingDeadlineJob);
                 }
              }
              System.out.println();
              invalidJobs.addAll(missingDeadlineJobs);
           }

           if (noWalltimeJobs.size() > 0) {
              System.out.print("Purge no estimated execution time jobs : ");
              for (String noWalltimeJob : noWalltimeJobs) {
                 System.out.print(noWalltimeJob + " ");
              }
              System.out.println();
              invalidJobs.addAll(noWalltimeJobs);
           }

           if (unconfirmedJobs.size() > 0) {
              System.out.print("Purge unconfirmed jobs : ");
              for (String unconfirmedJob : unconfirmedJobs) {
                 System.out.print(unconfirmedJob + " ");
                 Connection dbConn = getDbConnection();
                 Statement dbStmt = dbConn.createStatement();
                 if (dbStmt != null) {
                    dbStmt.execute("update jobinfo set jobstate='CONFIRMATION_EXPIRED' where pbsjobid=" + unconfirmedJob);
                 }
              }
              dbConn.close();
              System.out.println();
              invalidJobs.addAll(unconfirmedJobs);
           }

           if (invalidJobs.size() > 0) {
              try {
                 admissionOperationMgr.reschedule(this);
              } catch (Exception e) {
                 e.printStackTrace(); // No server to connect to (15034)
              }
           }

        }

        public String qcommand() { // callback method
           try {
              if ( (invalidJobs == null) || (invalidJobs.size() == 0) ) return "";
              String cmd = "qdel ";
              for (String invalidJob : invalidJobs) {
                 cmd += invalidJob + "[] ";
              }
              
              Process p = Runtime.getRuntime().exec(cmd);
              BufferedReader stdInput = new BufferedReader(new InputStreamReader(p.getInputStream()));

              BufferedReader stdError = new BufferedReader(new InputStreamReader(p.getErrorStream()));

              // read the output from the command
              //System.out.println("Here is the standard output of the command:\n");
              String s = null;
              while ((s = stdInput.readLine()) != null) {
                 System.out.println(s);
              }

             // read any errors from the attempted command
              //System.out.println("Here is the standard error of the command (if any):\n");
              boolean hasError = false;
              while ((s = stdError.readLine()) != null) {
                 System.err.println(s);
                 hasError = true;
              }
              if (hasError) throw new RuntimeException(s);

           }
           catch (Exception e) {
              e.printStackTrace();
              //throw new RuntimeException(e);
           }

           return "";
        }

        public String qcommand(String argument) { // callback method
           throw new RuntimeException("Arbitrary Qcommand not supported");
        }

}
